package com.zhanglijie.improve.collections.delayqueue;

import java.awt.geom.NoninvertibleTransformException;
import java.util.AbstractQueue;
import java.util.Collection;
import java.util.Collections;
import java.util.Iterator;
import java.util.PriorityQueue;
import java.util.Spliterator;
import java.util.concurrent.BlockingDeque;
import java.util.concurrent.Delayed;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;
import java.util.function.Consumer;
import java.util.function.Predicate;
import java.util.logging.Level;
import java.util.stream.Stream;

import static java.util.concurrent.TimeUnit.NANOSECONDS;

/**
 * @author zhanglijie
 * @version 1.0
 * @since 1.1.0 2022/12/4 0004 17:09
 */
public class MyDelayQueue<E extends Delayed> extends AbstractQueue<E> implements BlockingDeque<E> {

    //非公平锁
    private final transient ReentrantLock lock = new ReentrantLock();
    //使用priorityQueue存储元素 实现最小堆
    private final PriorityQueue<E> q = new PriorityQueue<>();
    //用来阻塞出队线程
    private final Condition available = lock.newCondition();
    //leader-follower 模型下的leader标记（他总是处理队首元素）
    private Thread leader = null;
    /**
     * 初始化一个空的延迟队列
     */
    public MyDelayQueue(){

    }

    /**
     * 以一个集合构造延时队列
     * @param c
     */
    public MyDelayQueue(Collection<? extends E> c){
        this.addAll(c);
    }

    /**
     * 删除一个元素
     * @param o
     * @return
     */
    @Override
    public boolean remove(Object o) {
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            return q.remove(o);
        } finally {
            lock.unlock();
        }
    }

    /**
     * 剩余容量--小根堆会自动扩容的
     * @return
     */
    @Override
    public int remainingCapacity() {
        return Integer.MAX_VALUE;
    }


    /**
     * 清空
     */
    @Override
    public void clear() {
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            q.clear();
        } finally {
            lock.unlock();
        }
    }


    /**
     * 讲队列的元素（到期的）排除到队列c中
     * @param c
     * @return
     */
    @Override
    public int drainTo(Collection<? super E> c) {
        if (c == null)
            throw new NullPointerException();
        if (c == this)
            throw new IllegalArgumentException();
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            int n = 0;
            //为啥判断到第一个未到期的就不要往下添加的，因为这个是按照delay时间排序的小根堆
            for (E e; (e = peekExpired()) != null;) {
                c.add(e);       // In this order, in case add() throws.
                q.poll();
                ++n;
            }
            return n;
        } finally {
            lock.unlock();
        }
    }
    /**
     * 对头元素过期才返回的get
     * 为啥这个方法不加锁，因为这个方法是私有方法，它会被本类的public方法调用，
     * 那么加锁逻辑必定要放在public方法中
     * @return
     */
    private E peekExpired() {
        // assert lock.isHeldByCurrentThread();
        E first = q.peek();
        return (first == null || first.getDelay(NANOSECONDS) > 0) ?
            null : first;
    }


    @Override
    public int size() {
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            return q.size();
        } finally {
            lock.unlock();
        }
    }

    /**
     * 带中断阻塞和超时的get
     * @param timeout
     * @param unit
     * @return
     * @throws InterruptedException
     */
    @Override
    public E poll(long timeout, TimeUnit unit) throws InterruptedException {
        long nanos = unit.toNanos(timeout);
        final ReentrantLock lock = this.lock;
        lock.lockInterruptibly();
        try{
            for(;;){
                E first = q.peek();
                if(first == null){
                    //由于是给了超时参数
                    if(nanos <= 0){
                       return null;
                    }
                    else{
                        //被唤醒的剩余阻塞时间
                        nanos = available.awaitNanos(nanos);
                    }
                }else{//队列不为空
                    long delay = first.getDelay(NANOSECONDS);
                    //判断队首元素是否已经到期
                    if(delay <= 0){
                        return q.poll();
                    }
                    //delay时间没有到,但是剩余时间也没有了，只能返回null
                    if(nanos <= 0){
                        return null;
                    }
                    //防止内存泄露
                    first = null;
                    /**
                     * 走到这里表示：元素时间每到 等待时间也是有的
                     * 1.nanoc<delay 说明当前线程肯定等不到队首元素了
                     * 2.nanocs>=delay 但是leader!=null 说明当前线程能等到对头元素
                     *   2.1但是有leader了，那么只能让leader来唤醒自己,不能获得
                     *   2.2 可以阻塞delay时间获得
                     *
                     */
                    if(nanos < delay || leader != null){
                        //nano<delay还是要等的 因为nanos时间内又加入个delay时间更小的元素
                        //小到当前线程获得新队首元素出队
                        nanos = available.awaitNanos(nanos);
                    }else{//nanos >= delay 且 leader==null
                        Thread currentThread = Thread.currentThread();
                        leader = currentThread;
                        try{

                            long timeLeft = available.awaitNanos(delay);
                            //下次需要等待的时间 = nanos-此次消耗的时间（因为可能被别人notify)
                            nanos -=delay-timeLeft;
                        }finally {
                            if(leader == currentThread){
                                leader = null;
                            }
                        }
                    }
                }
            }
        }finally {
            //这个finally是在return 前执行的 先E temp = p.poll()->finally->return temp; 所以finally判断q.peek()是temp后面的元素除了temp还有没有元素
            if(leader == null && q.peek() != null){
                available.signal();
            }
            lock.unlock();
        }
    }


    @Override
    public E peek() {
        final ReentrantLock lock = this.lock;
        lock.lock();
        try{
            return q.peek();
        }finally {
            lock.unlock();
        }
    }
    /**
     * 不阻塞的get元素
     * @return
     */
    @Override
    public E poll() {
        final ReentrantLock lock = this.lock;
        lock.lock();
        try{
            E first = q.peek();
            if(first == null || first.getDelay(NANOSECONDS) > 0){
                return null;
            }
            //否则直接出队
            else{
                return q.poll();
            }

        }finally {
            lock.unlock();
        }
    }

    /**
     * 注意先执行finally再执行return的
     * 会阻塞get
     * @return
     * @throws InterruptedException
     */
    @Override
    public E take() throws InterruptedException {
        final ReentrantLock lock = this.lock;
        //必须加相应中断的锁
        lock.lockInterruptibly();
        try{
            for(;;){
                E first = q.peek();
                if(first == null){
                    //阻塞
                   available.await();
                }else{//队列不为空
                    //获取队列的延时时间
                    long delay = first.getDelay(NANOSECONDS);
                    //时间到了直接出队
                    if(delay <= 0){
                        return q.poll();
                    }
                    first = null;//如果很多线程都拿first 但是没资格拿走 first后面被拿走了 但是n-1个线程还没释放first会内存溢出
                    //走到这里队首元素未到期
                    if(leader != null){
                        available.await();
                    }
                    //走到这里 表示delay时间没有到，又没有线程是leader 那只有自己作为leader了
                    //正常第一个线程会走到这里
                    else{
                        Thread currentThread = Thread.currentThread();
                        leader = currentThread;
                        try{
                            available.awaitNanos(delay);
                        }finally {//finally要不是中断释放leader 要不是正常醒来的时候释放leader
                            if(leader == currentThread){
                                leader = null;
                            }
                        }
                    }
                }
            }
        }finally {
            //这个finally是在return 前执行的 先E temp = p.poll()->finally->return temp; 所以finally判断q.peek()是temp后面的元素除了temp还有没有元素
            if(leader == null && q.peek() != null){
                available.signal();
            }
            lock.unlock();
        }
    }
    /**
     * 这个offer的超时参数是无效的 不支持超时和中断
     * @param e
     * @param timeout
     * @param unit
     * @return
     * @throws InterruptedException
     */
    @Override
    public boolean offer(E e, long timeout, TimeUnit unit) throws InterruptedException {
        return offer(e);
    }
    @Override
    public void put(E e) throws InterruptedException {
        offer(e);
    }
    @Override
    public boolean add(E e){
        return offer(e);
    }
    @Override
    public boolean offer(E e) {
        final ReentrantLock lock = this.lock;
        lock.lock();
        try{
            //当前线程相最小堆添加元素
            q.offer(e);
            /**
             * 这里获取的队头==e,表示此时e是优先级最高的元素，说明之前的leader，他的awaitNonos的超时时间
             * 过大，此时有更小的进来 后面为了不被忽略，将会唤醒一个follwer作为新的leader,之前的leader
             * 到了时间进入worker队列中执行即可。
             */
            if(q.peek() == e){
                leader = null;
                available.signal();
            }
            return true;
        }finally {
            lock.unlock();
        }
    }

    @Override
    public void addFirst(E e) {

    }

    @Override
    public void addLast(E e) {

    }

    @Override
    public boolean offerFirst(E e) {
        return false;
    }

    @Override
    public boolean offerLast(E e) {
        return false;
    }

    @Override
    public E removeFirst() {
        return null;
    }

    @Override
    public E removeLast() {
        return null;
    }

    @Override
    public E pollFirst() {
        return null;
    }

    @Override
    public E pollLast() {
        return null;
    }

    @Override
    public E getFirst() {
        return null;
    }

    @Override
    public E getLast() {
        return null;
    }

    @Override
    public E peekFirst() {
        return null;
    }

    @Override
    public E peekLast() {
        return null;
    }

    @Override
    public void putFirst(E e) throws InterruptedException {

    }

    @Override
    public void putLast(E e) throws InterruptedException {

    }

    @Override
    public boolean offerFirst(E e, long timeout, TimeUnit unit) throws InterruptedException {
        return false;
    }

    @Override
    public boolean offerLast(E e, long timeout, TimeUnit unit) throws InterruptedException {
        return false;
    }

    @Override
    public E takeFirst() throws InterruptedException {
        return null;
    }

    @Override
    public E takeLast() throws InterruptedException {
        return null;
    }

    @Override
    public E pollFirst(long timeout, TimeUnit unit) throws InterruptedException {
        return null;
    }

    @Override
    public E pollLast(long timeout, TimeUnit unit) throws InterruptedException {
        return null;
    }

    @Override
    public boolean removeFirstOccurrence(Object o) {
        return false;
    }

    @Override
    public boolean removeLastOccurrence(Object o) {
        return false;
    }




    @Override
    public int drainTo(Collection<? super E> c, int maxElements) {
        return 0;
    }

    @Override
    public void push(E e) {

    }

    @Override
    public E pop() {
        return null;
    }

    @Override
    public Iterator<E> descendingIterator() {
        return null;
    }

    @Override
    public boolean removeIf(Predicate<? super E> filter) {
        return false;
    }

    @Override
    public Spliterator<E> spliterator() {
        return null;
    }

    @Override
    public Stream<E> stream() {
        return null;
    }

    @Override
    public Stream<E> parallelStream() {
        return null;
    }

    @Override
    public Iterator<E> iterator() {
        return null;
    }


}
